#include "nwtcpserver.h"
#include <chrono>
#include <pthread.h>


namespace NetworkUtil {

static std::function<void(const std::string& strDataBuf, NetworkUtil::Session* pSession)> gs_pfnRecvCallback = nullptr;


void setRecvCallbackFunc(std::function<void(const std::string& strDataBuf, NetworkUtil::Session* pSession)> pfnRecvCallback)
{
    gs_pfnRecvCallback = pfnRecvCallback;
    return;
}


Session::Session(uint32_t ui32BufMaxSize
                 , asio::ip::tcp::socket socket
                 , NetworkUtil::TcpServer* pSrvHandler)
    : m_ui32BufMaxSize(ui32BufMaxSize)
    , m_objSocket(std::move(socket))
    , m_pSrvHandler(pSrvHandler)
{

}

Session::~Session()
{

}

void Session::start()
{
    //std::cout << "Session::start..." << std::endl;
    if (!m_bRunning)
    {
        m_bRunning = true;
        m_bConnected = true;

        if (!(m_sp4thRecv.get()))
        {
            m_sp4thRecv = std::make_shared<std::thread> (&Session::onRecvDataProcess, this);
        }
    }

    return;
}

void Session::stop()
{
    if (m_bRunning)
    {
        m_bRunning = false;
        m_bConnected = false;

        try {
            if (m_objSocket.is_open())
            {
                m_objSocket.shutdown(asio::ip::udp::socket::shutdown_both);
                m_objSocket.release();
                m_objSocket.close();
            }

            if (m_sp4thRecv.get())
            {
                if (m_sp4thRecv->joinable()) {
                    m_sp4thRecv->join();
                }
                m_sp4thRecv.reset();
                m_sp4thRecv = nullptr;
            }
        } catch (...) {
            // Exception
        }
    }

    return;
}

uint32_t Session::read(char* pDataBuf, uint32_t ui32DataBufLen)
{
    uint32_t ui32Ret = 0;
    try {
        if (m_objSocket.is_open() && pDataBuf)
        {
            std::size_t s32Ret = m_objSocket.read_some(asio::buffer(pDataBuf, static_cast<std::size_t>(ui32DataBufLen)));
            ui32Ret = static_cast<uint32_t>(s32Ret);
        }
    } catch (std::exception& e) {
        std::cerr << "func [Session::read], exception info: " << e.what() << std::endl;
        clientLostConnection();
    }

    return ui32Ret;
}

uint32_t Session::read(std::string& strDataBuf)
{
    char *pBuffer = new (std::nothrow) char[m_ui32BufMaxSize];
    if (!pBuffer)
    {
        return 0;
    }

    memset(pBuffer, 0, m_ui32BufMaxSize);
    uint32_t ui32Ret = read(pBuffer, m_ui32BufMaxSize);
    if (0 < ui32Ret)
    {
        strDataBuf = std::string(pBuffer, ui32Ret);
    }

    if (pBuffer)
    {
        delete[] pBuffer;
        pBuffer = nullptr;
    }

    return ui32Ret;
}

uint32_t Session::write(const char* pDataBuf, uint32_t ui32DataBufLen)
{
    uint32_t ui32Ret = 0;
    try {
        if (m_objSocket.is_open() && pDataBuf)
        {
            std::size_t s32Ret = asio::write(m_objSocket
                                             , asio::buffer(pDataBuf, static_cast<std::size_t>(ui32DataBufLen)));
            ui32Ret = static_cast<uint32_t>(s32Ret);
        }
    } catch (std::exception& e) {
        std::cerr << "func [Session::write], exception info: " << e.what() << std::endl;
        clientLostConnection();
    }

    return ui32Ret;
}

uint32_t Session::write(const std::string& strDataBuf)
{
    uint32_t ui32Ret = 0;

    try {
        if (m_objSocket.is_open())
        {
            std::size_t s32Ret = m_objSocket.write_some(asio::buffer(strDataBuf));
            ui32Ret = static_cast<uint32_t>(s32Ret);
        }
    } catch (std::exception& e) {
        std::cerr << "func [Session::write], exception info: " << e.what() << std::endl;
        clientLostConnection();
    }

    return ui32Ret;
}

std::string Session::remoteInfo()
{
    std::string strInfo = "";
    if (m_objSocket.is_open())
    {
        strInfo = m_objSocket.remote_endpoint().address().to_string() + ":" + std::to_string(m_objSocket.remote_endpoint().port());
    }

    return strInfo;
}

void Session::onRecvDataProcess()
{
    while (m_bRunning)
    {
        if (m_bConnected)
        {
            std::string strDataInfo = "";
            read(strDataInfo);

            if (!(strDataInfo.empty()))
            {
                if (gs_pfnRecvCallback)
                {
                    gs_pfnRecvCallback(strDataInfo, this);
                }
            }
        }

        std::this_thread::yield();
    }

    return;
}

void Session::clientLostConnection()
{
    m_bConnected = false;

    std::thread([this]()
    {
        if (this->m_pSrvHandler)
        {
            this->m_pSrvHandler->removeClient(this);
        }
    }).detach();

    return;
}


TcpServer::TcpServer(const ConnectionConfigInfo& stConnCfgInfo)
    : BaseConnection(stConnCfgInfo)
{

}

TcpServer::~TcpServer()
{
}

void TcpServer::init()
{
    if (m_bInited)
    {
        return;
    }

    try {

        m_mt4Sessions.lock();
        m_vt4spSessions.clear();
        m_mt4Sessions.unlock();

        if (!(m_sp4ObjIOCtx.get()))
        {
            m_sp4ObjIOCtx = std::make_shared<asio::io_context>();
        }

        if (!(m_sp4ObjAcceptor.get()))
        {
            m_sp4ObjAcceptor = std::make_shared<asio::ip::tcp::acceptor>(*(m_sp4ObjIOCtx.get())
                                                                         , asio::ip::tcp::endpoint(asio::ip::tcp::v4()
                                                                                                   , m_stConnCfgInfo.ui16LocalPort));
            //m_sp4ObjAcceptor->listen();
        }

        m_bRunning = true;

        if (m_stConnCfgInfo.bAsync) {
            // 阻塞线程，退出失败
            onAsyncAcceptBusinessProcess();
        } else {
            if (!(m_sp4thAcceptBusiness.get()))
            {
                m_sp4thAcceptBusiness = std::make_shared<std::thread> (&TcpServer::onSyncAcceptBusinessProcess, this);
            }
        }

        m_sp4ObjIOCtx->run();
        m_bInited = true;

    } catch (std::exception& e) {
        printf("func [TcpServer::init], start server (%d) failed, error info: %s.\r\n" \
               , m_stConnCfgInfo.ui16LocalPort, e.what());

        m_bInited = true;
        deinit();
    }

    return;
}

void TcpServer::deinit()
{
    if (m_bInited)
    {
        try {
            if (m_sp4ObjAcceptor.get())
            {
                m_sp4ObjAcceptor->cancel();
                m_sp4ObjAcceptor->release();

                if (m_sp4ObjAcceptor->is_open()) {
                    asio::error_code ignored_ec;
                    m_sp4ObjAcceptor->close(ignored_ec);
                }

                m_sp4ObjAcceptor.reset();
                m_sp4ObjAcceptor = nullptr;
            }

            if (m_sp4ObjIOCtx.get())
            {
                m_sp4ObjIOCtx->stop();
                m_sp4ObjIOCtx.reset();
                m_sp4ObjIOCtx = nullptr;
            }
        } catch (...) {
            // Exception
        }

        m_bRunning = false;
        if (m_sp4thAcceptBusiness.get())
        {
            pthread_cancel(m_sp4thAcceptBusiness->native_handle()); // 因为acceptor阻塞线程，所以需要强行杀死这个线程
            if (m_sp4thAcceptBusiness->joinable()) {
                m_sp4thAcceptBusiness->join();
            }
            m_sp4thAcceptBusiness.reset();
            m_sp4thAcceptBusiness = nullptr;
        }

        m_mt4Sessions.lock();
        for (size_t i = 0, size = m_vt4spSessions.size(); i < size; ++i)
        {
            if (m_vt4spSessions[i].get())
            {
                m_vt4spSessions[i]->stop();
                m_vt4spSessions[i].reset();
                m_vt4spSessions[i] = nullptr;
            }
        }
        m_vt4spSessions.clear();
        m_mt4Sessions.unlock();

        m_bInited = false;
    }

    return;
}

void TcpServer::removeClient(NetworkUtil::Session* pSession)
{
    if (!pSession)
    {
        return;
    }

    m_mt4Sessions.lock();
    for (auto iter = m_vt4spSessions.begin(); iter != m_vt4spSessions.end(); ++iter)
    {
        if (iter->get()  == pSession)
        {
            pSession->stop();
            iter = m_vt4spSessions.erase(iter);
            pSession = nullptr;
            break;
        }
    }
    m_mt4Sessions.unlock();

    return;
}

void TcpServer::onAsyncAcceptBusinessProcess()
{
    if (m_sp4ObjAcceptor.get()) {
        m_sp4ObjAcceptor->async_accept([this](std::error_code ec, asio::ip::tcp::socket socket)
        {
            if (!ec)
            {
                std::shared_ptr<NetworkUtil::Session> sp4Session = std::make_shared<NetworkUtil::Session>(m_stConnCfgInfo.ui32BufMaxSize
                                                                                                          , std::move(socket), this);
                if (sp4Session.get())
                {
                    //std::cout << "connect: " << sp4Session->remoteInfo() << std::endl;
                    sp4Session->start();

                    m_mt4Sessions.lock();
                    m_vt4spSessions.emplace_back(sp4Session);
                    m_mt4Sessions.unlock();
                }
            }

            onAsyncAcceptBusinessProcess();
        });
    }

    return;
}

void TcpServer::onSyncAcceptBusinessProcess()
{
    while (m_bRunning)
    {
        if (m_sp4ObjAcceptor.get())
        {
            // 同步阻塞，线程退出会失败
            std::shared_ptr<NetworkUtil::Session> sp4Session = std::make_shared<NetworkUtil::Session>(m_stConnCfgInfo.ui32BufMaxSize
                                                                                                      , m_sp4ObjAcceptor->accept()
                                                                                                      , this);
            if (sp4Session.get())
            {
                //std::cout << "connect: " << sp4Session->remoteInfo() << std::endl;
                sp4Session->start();

                m_mt4Sessions.lock();
                m_vt4spSessions.emplace_back(sp4Session);
                m_mt4Sessions.unlock();
            }
        }

        std::this_thread::yield();
    }

    return;
}

} // namespace NetworkUtil
